package org.yonggan.dmp.utils


import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.Row
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer


/**
  * 用户标签提取工具
  *
  */
object UserTagUtils {

  private val DEFUALT_TUP = (("", 0))


  /**
    * 提取用户表示信息
    *
    * @param row
    */
  def userIdentity(row: Row): String = {

    row match {
      case v if StringUtils.isNotEmpty(row.getAs[String]("imei")) => "IM#" + row.getAs[String]("imei")
    case v if StringUtils.isNotEmpty(row.getAs[String]("mac")) => "MC#" + row.getAs[String]("mac")
    case v if StringUtils.isNotEmpty(row.getAs[String]("idfa")) => "IF#" + row.getAs[String]("idfa")
    case v if StringUtils.isNotEmpty(row.getAs[String]("openudid")) => "PU#" + row.getAs[String]("openudid")
    case v if StringUtils.isNotEmpty(row.getAs[String]("androidid")) => "AD#" + row.getAs[String]("androidid")

    case v if StringUtils.isNotEmpty(row.getAs[String]("imeimd5")) => "IMM#" + row.getAs[String]("imeimd5")
    case v if StringUtils.isNotEmpty(row.getAs[String]("macmd5")) => "MCM#" + row.getAs[String]("macmd5")
    case v if StringUtils.isNotEmpty(row.getAs[String]("idfamd5")) => "IFM#" + row.getAs[String]("idfamd5")
    case v if StringUtils.isNotEmpty(row.getAs[String]("openudidmd5")) => "PU#" + row.getAs[String]("openudidmd5")
    case v if StringUtils.isNotEmpty(row.getAs[String]("androididmd5")) => "ADM#" + row.getAs[String]("androididmd5")

    case v if StringUtils.isNotEmpty(row.getAs[String]("imeisha1")) => "IMS#" + row.getAs[String]("imeisha1")
    case v if StringUtils.isNotEmpty(row.getAs[String]("macsha1")) => "MCS#" + row.getAs[String]("macsha1")
    case v if StringUtils.isNotEmpty(row.getAs[String]("idfasha1")) => "IFS#" + row.getAs[String]("idfasha1")
    case v if StringUtils.isNotEmpty(row.getAs[String]("openudidsha1")) => "PUS#" + row.getAs[String]("openudidsha1")
    case v if StringUtils.isNotEmpty(row.getAs[String]("androididsha1")) => "AD#" + row.getAs[String]("androididsha1")
    case v  => "未知用户"
  }

  }


  /**
    * 广告位类型
    * 标签格式： LC03->1 或者 LC16->1）xx 为数字，小于 10 补 0
    *
    * 位类型名称，LN 插屏->1
    *
    * @param adtype
    * @return  List(at,an)  at:广告位类型 an:广告位类型名称
    *
    */
  def adSpaceType(adtype: String, adname: String): List[(String, Int)] = {
    // 广告位类型
    val at = if (StringUtils.isNoneEmpty(adtype)) ("LC" + numAddzero(adtype), 1) else DEFUALT_TUP
    // 广告位类型名称
    val an = if (StringUtils.isNoneEmpty(adname)) ("LN" + adname, 1) else DEFUALT_TUP
    List(at, an)
  }

  /**
    * 数字补零操作
    *
    * @param num
    * @return
    */
  def numAddzero(num: String) = if (Integer.valueOf(num) < 10) "0" + num else num


  /**
    * 提取Appname
    * 14 appid: String, 应用 id 15 appname: String, 应用名称
    *
    * @param row
    * @param jedis redis 连接
    */
  def getAppName(row: Row, jedis: Jedis): List[(String, Int)] = {

    var lst = List[(String, Int)]()

    var appname = row.getAs[String]("appname")
    val appid = row.getAs[String]("appname")
    //  提取appName
    if (StringUtils.isEmpty(appname)) {
      if (StringUtils.isNotEmpty(appid)) {
        appname = jedis.hget("app_dmp", "N-" + appid)
      } else appname = appid
      lst :+= (appname, 1)
    }
    lst
  }


  /**
    * 提取用户设备详细信息
    *
    * a) (操作系统 -> 1)
    * b) (联网方 -> 1)
    * c) (运营商 -> 1)
    *
    * 设备操作系统
    * 1 Android D00010001
    * 2 IOS D00010002
    * 3 WinPhone D00010003 _
    * 其他 D00010004
    *
    * @return List
    */
  def equipmentInfo(row: Row): List[(String, Int)] = {
    //    操作系统
    //    client: Int, 设备类型 （1：android 2：ios 3：wp
    val client = row.getAs[Int]("client")
    match {
      case 1 => ("D00010001", 1)
      case 2 => ("D00010002", 1)
      case 4 => ("D00010003", 1)
      case _ => ("D00010004", 1)
    }

    //    联网方式名称
    val netname = row.getAs[String]("networkmannername")
    match {
      case "Wifi" => ("D00020001", 1)
      case "4G" => ("D00020002", 1)
      case "3G" => ("D00020003", 1)
      case "2G" => ("D00020004", 1)
      case _ => ("D00020005", 1)
    }

    // 运营商名称
    val ispname = row.getAs[String]("ispname")
    match {
      case "移动" => ("D00030001", 1)
      case "联通" => ("D00030002", 1)
      case "电信" => ("D00030003", 1)
      case _ => ("D00030004", 1)
    }
    List(client, netname, ispname)
  }

  /**
    * 地域标签
    *
    */
  def areaTag(row: Row): List[(String, Int)] = {
    val pname = row.getAs[String]("provincename")
    val cname = row.getAs[String]("cityname")
    List(("ZP" + pname, 1), ("ZC" + cname, 1))
  }

  /**
    * 提取关键词标签
    * 复仇|热门电影|动作|类型|惊悚|题材|犯罪|政治|地区|情感|类型|当代|时代|救援
    *
    * @param row
    */
  def keyWordTag(row: Row, stopWordBct: Broadcast[Array[String]]): List[(String, Int)] = {
    val keywords = row.getAs[String]("keywords")
    val splitArr = keywords.split("\\|", -1)

    // 停用词集合
    val stopWordColl = stopWordBct.value.toList

    var keywordsLst = new ListBuffer[(String, Int)]
    for (keyWordTemp: String <- splitArr) {
      // 停用词和关键词过滤
      if (!stopWordColl.contains(keyWordTemp)
        && keyWordTemp.length > 3 && keyWordTemp.length < 8) {
        keywordsLst.:+=((keyWordTemp, 1))
      }
    }
    keywordsLst.toList
  }
}
